package com.tid.hbasees.utils.hbase;

import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.tid.hbasees.SysConfig;
import com.tid.hbasees.utils.JsonHelper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

@Component("hbaseHelper")
public class HBaseHelper {
    private static final Logger LOG = Logger.getLogger(HBaseHelper.class);

    public static String ROWKEY_SEP = ",#,";

    private ExecutorService pool;

    @Resource
    private SysConfig sysConfig;

    @Resource
    private JsonHelper jsonHelper;

    private HConnection conn;

    @PostConstruct
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.ZOOKEEPER_QUORUM, sysConfig.getHBaseZks());
        this.conn = HConnectionManager.createConnection(conf);
        this.pool = Executors.newFixedThreadPool(20);
    }

    public HTableInterface getTable(String tableName) throws IOException {
        return this.conn.getTable(tableName);
    }

    public String getStringValue(Result r, String qualifier) {
        byte[] v = r.getValue(Bytes.toBytes("cf"), Bytes.toBytes(qualifier));
        return v != null ? Bytes.toString(v) : null;
    }

    public List<String> indexScanResult(Map<String, Scan> scans) throws ExecutionException, InterruptedException {
        List<Future<List<String>>> futures = Lists.newLinkedList();
        for (Map.Entry<String, Scan> s : scans.entrySet()) {
            futures.add(pool.submit(new ScanWorker(s.getValue(), s.getKey(),
                    this)));
        }

        Set<String> rowkeys = Sets.newLinkedHashSet();

        for (Future<List<String>> f : futures) {
            rowkeys.addAll(f.get());
        }

        return Lists.newLinkedList(rowkeys);
    }

    public List<String> indexScanResult(List<Map<String, Scan>> scans) throws ExecutionException, InterruptedException {
        List<Future<List<String>>> futures = Lists.newLinkedList();
        Set<String> result = Sets.newLinkedHashSet();
        for (Map<String, Scan> scan : scans) {
            for (Map.Entry<String, Scan> s : scan.entrySet()) {
                futures.add(pool.submit(new ScanWorker(s.getValue(), s.getKey(),
                        this)));
            }

            Set<String> rowkeys = Sets.newLinkedHashSet();

            for (Future<List<String>> f : futures) {
                rowkeys.addAll(f.get());
            }

            if (result.isEmpty()) {
                result.addAll(rowkeys);
            } else {
                Set<String> s = Sets.intersection(result, rowkeys);
                result.clear();
                result.addAll(s);
            }
        }



        return Lists.newLinkedList(result);
    }

    public <T> List<T> multiScan(Map<String, Scan> scans, String tableName, Class<T> clazz)
            throws ExecutionException, InterruptedException {
        return multiScan(Lists.newArrayList(scans), tableName, clazz);
    }

    public <T> List<T> multiScan(List<Map<String, Scan>> scans, String tableName, Class<T> clazz)
            throws ExecutionException, InterruptedException {

        List<Get> gets = Lists.newLinkedList();
        for (String rowkey : indexScanResult(scans)) {
            gets.add(new Get(Bytes.toBytes(rowkey)));
        }

        List<T> result = Lists.newLinkedList();
        try (HTableInterface table = getTable(tableName)) {
            for (Result r : table.get(gets)) {
                String json = getJsonContent(r);
                if (json != null) {
                    Optional<T> nodeOptional = jsonHelper.readValue(json, clazz);
                    if (nodeOptional.isPresent()) {
                        result.add(nodeOptional.get());
                    }
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            LOG.error("failed to scan table " + tableName);
        }

        return result;
    }

    public Scan buildTypicalScan(String id, long startTime, long endTime) {
        Scan s = new Scan();
        if (startTime > 0 && endTime > 0 && endTime > startTime) {
            byte[] start = Bytes.toBytes(id + HBaseHelper.ROWKEY_SEP + startTime);
            byte[] stop = Bytes.toBytes(id + HBaseHelper.ROWKEY_SEP + endTime);
            s.setStartRow(start);
            s.setStopRow(stop);
        } else {
            byte[] start = Bytes.toBytes(id);
            byte[] stop = start.clone();
            stop[stop.length - 1] = (byte) (stop[stop.length - 1] + 1);
            s.setStartRow(start);
            s.setStopRow(stop);

        }

        return s;
    }

    public Scan buildTypicalScan(long startTime, long endTime) {
        Scan s = new Scan();
        if (startTime > 0 && endTime > 0 && endTime > startTime) {
            byte[] start = Bytes.toBytes(startTime);
            byte[] stop = Bytes.toBytes(endTime);
            s.setStartRow(start);
            s.setStopRow(stop);
        }
        return s;
    }

    public String getJsonContent(Result r) {
        return getStringValue(r, "json");
    }
}
